package com.github.ghsea.cc.bolt;

import java.util.Date;
import java.util.Map;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.github.ghsea.cc.consts.FiledName;
import com.github.ghsea.cc.service.analysis.Statistics;
import com.github.ghsea.cc.util.SpringUtil;

/**
 * 根据IP计算access log
 * 
 * @author GuHai 2017-7-27下午10:39:08
 */
public class CounterByIpBolt extends BaseBasicBolt {

	private Logger log = LoggerFactory.getLogger(NginxLogSplitBolt.class);

	private static final long serialVersionUID = 8670248862245526549L;

	private Statistics statics = null;

	public void execute(Tuple input, BasicOutputCollector collector) {
		String requestUrl = input.getStringByField(FiledName.REQUEST_URL);
		String requestHost = input.getStringByField(FiledName.REQUEST_HOST);
		String accessIp = input.getStringByField(FiledName.ACCESS_IP);
		Date accessTime = (Date) input.getValueByField(FiledName.ACCESS_TIME);

		log.info("CounterByIpBolt received:requestUrl=" + requestUrl + ",accessIp=" + accessIp + ",accessTime="
				+ accessTime + ",requestHost=" + requestHost);

		statics.incrTimesAndCheckCC(accessIp, requestHost, requestUrl, accessTime);
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub

	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		statics = SpringUtil.getBean(Statistics.class);
	}

}
